package com.yang.config;

import com.aliyun.openservices.ons.api.*;
import com.yang.service.impl.FireTopicMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * 阿里云RocketMQ配置类
 */
@Configuration
@Slf4j
public class RocketMQConfig {

    @Autowired
    private AliRocketMQTopicProperties aliRocketMQTopicProperties;

    /**
     * 创建生产者
     * @return
     */
    @Bean
    public Producer producer() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "GID_publicmet_response");
        properties.put(PropertyKeyConst.AccessKey, "su67gCQTyUXOscZE");
        properties.put(PropertyKeyConst.SecretKey, "Tul8EJr293DBCd1b");
        // 专网访问
//        properties.put(PropertyKeyConst.NAMESRV_ADDR, "ep-bp1idd7c6197a104a564.epsrv-bp1g2k84c7kid2igyiun.cn-hangzhou.privatelink.aliyuncs.com:8080");
        // 公网访问
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "rmq-cn-36z3zyznz04.cn-hangzhou.rmq.aliyuncs.com:8080");
        properties.put(PropertyKeyConst.Namespace, "rmq-cn-36z3zyznz04");
        Producer producer = ONSFactory.createProducer(properties);
        producer.start();
        log.info("生产者创建成功");
        return producer;
    }

    /**
     * 创建消费者
     * @return
     */
    @Bean
    public Consumer consumer() {
        Properties properties = new Properties();
        // 设置消费者Group ID，需要在阿里云控制台创建
        properties.put(PropertyKeyConst.GROUP_ID, "GID_publicmet_response");
        // AccessKey ID 和 AccessKey Secret
        properties.put(PropertyKeyConst.AccessKey, "su67gCQTyUXOscZE");
        properties.put(PropertyKeyConst.SecretKey, "Tul8EJr293DBCd1b");
        // 设置NameServer地址
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "rmq-cn-36z3zyznz04.cn-hangzhou.rmq.aliyuncs.com:8080");
        // 设置实例ID
        properties.put(PropertyKeyConst.Namespace, "rmq-cn-36z3zyznz04");
        Consumer consumer = ONSFactory.createConsumer(properties);

        // 订阅一个或多个Topic，以及Tag来过滤需要消费的消息
//        consumer.subscribe("topic", "tag", new MessageListener() {
        // tag设置什么类型就会消费什么类型，如果设置为""，那么就不会根据tag标签进行过滤，有消息就会进行消费
//        consumer.subscribe(aliRocketMQTopicProperties.getFireTest(), "", new MessageListener() {
//            @Override
//            public Action consume(Message message, ConsumeContext consumeContext) {
//                log.info("Receive message:{}", new String(message.getBody()));
//                String receiveMessage = new String(message.getBody());
//                SatelliteRemoteSensingFireData satelliteRemoteSensingFireData = JSON.parseObject(receiveMessage, SatelliteRemoteSensingFireData.class);
//                log.info("转换为对象后：{}", JSON.toJSONString(satelliteRemoteSensingFireData));
//                // 返回Action.CommitMessage表示消费成功，消息会被删除
//                // 返回Action.ReconsumeLater表示消费失败，消息稍后会被重新投递
//                return Action.CommitMessage;
//            }
//        });

        consumer.subscribe(aliRocketMQTopicProperties.getFireTest(), "", new FireTopicMessageListener());

        // 如果需要订阅多个Topic，再次订阅即可
//        consumer.subscribe();

        // 启动消费者实例
        consumer.start();

        log.info("消费者创建成功");
        return consumer;
    }

}
